package com.atguigu.flink.chapter07_state;

import com.atguigu.flink.pojo.WaterSensor;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/10/28
 *
 *      有两个流：
 *                  事实流
 *                      connect,关联信息。
 *                  维度流，每次更新信息时，可以把更新的信息广播到所有的事实流
 */
public class Demo4_BroadCastDemo
{



    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        //事实流
        SingleOutputStreamOperator<WaterSensor> factDs = env
            .socketTextStream("hadoop103", 8888)
            .map(new MapFunction<String, WaterSensor>()
            {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(
                        data[0],
                        Long.valueOf(data[1]),
                        Integer.valueOf(data[2])
                    );
                }
            });

        //维度流
        SingleOutputStreamOperator<DimBean> dimDs = env
            .socketTextStream("hadoop103", 8887)
            .map(new MapFunction<String, DimBean>()
            {
                @Override
                public DimBean map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new DimBean(
                        data[0], data[1]);
                }
            });

        //希望使用广播状态，要广播的状态所在的流必须是广播流
        //把dimDs变为广播流
        MapStateDescriptor<String, DimBean> mapStateDescriptor = new MapStateDescriptor<>("dim", String.class, DimBean.class);
        BroadcastStream<DimBean> broadcastDs = dimDs.broadcast(mapStateDescriptor);

        //事实流connect broadcastDs
        factDs.connect(broadcastDs)
              .process(new BroadcastProcessFunction<WaterSensor, DimBean, WaterSensor>()
              {

                  //处理事实流中的数据，事实流中的事实来一条调用一次
                  @Override
                  public void processElement(WaterSensor value, ReadOnlyContext ctx, Collector<WaterSensor> out) throws Exception {

                      ReadOnlyBroadcastState<String, DimBean> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

                      //获取广播后的状态中的维度信息
                      DimBean dimBean = broadcastState.get(value.getId());

                      if (dimBean != null){

                          value.setId(dimBean.getName());
                      }

                      out.collect(value);
                  }

                  /*
                        处理广播流，广播流中的数据来一条，把状态广播到所有的并行度。来一条数据，一个Process的并行度调一次
                   */
                  @Override
                  public void processBroadcastElement(DimBean value, Context ctx, Collector<WaterSensor> out) throws Exception {
                      System.out.println("Demo4_BroadCastDemo.processBroadcastElement");
                      //获取状态
                      BroadcastState<String, DimBean> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

                      //把维度信息放入状态，自动广播
                      broadcastState.put(value.getId(),value);

                  }
              })
              .print();

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    private static class DimBean{
        private String id;
        private String name;

    }
}
